package com.imooc.producer.b_order.test;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.utils.Const;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;

import java.util.List;

/**
 * @author 折戟沉沙铁未销
 * @version V1.0
 * @date 2024/4/10-2024
 * @Description: 分区顺序消费
 */
public class PartitionConsumerTestException {
    public static void main(String[] args) throws Exception {
        //定义consumer
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("partition_consumer_test");
        consumer.setNamesrvAddr(Const.NAME_SERVER_SINGLE_LOCAL);
        consumer.setMessageModel(MessageModel.CLUSTERING);

        consumer.subscribe("partition_orderly_topic","*");
        consumer.setMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                try {
                    System.out.println(JSONObject.toJSONString(msgs));
                    for (MessageExt msg : msgs) {
                        System.err.println(new String(msg.getBody()));
                        String tags = msg.getTags();
                        if(tags.equals("1002")){
                            int i = 1 / 0; /* 模拟一个运行异常*/
                        }
                    }
                    //确认这 msg 这一批消息
                    return ConsumeOrderlyStatus.SUCCESS;
                } catch (Exception e) {
                    e.printStackTrace();
                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                }
            }
        });

        //启动consumer
        consumer.start();


    }
}